package com.xbongbong.paas.uitl;

import com.xbongbong.paas.redis.PaasRedisHelper;
import com.xbongbong.paas.toolbox.util.StringUtil;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.concurrent.FutureCallback;
import org.apache.http.entity.ContentType;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.impl.nio.conn.PoolingNHttpClientConnectionManager;
import org.apache.http.impl.nio.reactor.DefaultConnectingIOReactor;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.nio.reactor.ConnectingIOReactor;
import org.apache.http.nio.reactor.IOReactorException;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Objects;


/**
 * 异步http连接工具类
 * @author kaka
 * @date  2021-7-2
 */
@Component
public class AsyncHttpClientUtil {

    private static final Logger LOG = LoggerFactory.getLogger(AsyncHttpClientUtil.class);

    private static Integer TRY_NUM = 3;

    private static Integer ERROR_NUM_SECONDS = 15 * 60;

    private static PaasRedisHelper staticPaasRedisHelper;

    private static CloseableHttpAsyncClient client;

    @Resource
    private PaasRedisHelper paasRedisHelper;

    @PostConstruct
    public void initClient() {
        staticPaasRedisHelper = paasRedisHelper;
        client = getHttpClient();
    }

    /**
     * 获取CloseableHttpAsyncClient单例实体
     *
     * @return
     */
    private static CloseableHttpAsyncClient getHttpClient() {
        if (client == null) {
            synchronized (AsyncHttpClientUtil.class) {
                if (client == null) {
                    RequestConfig requestConfig = RequestConfig.custom()
                            //连接超时,连接建立时间,三次握手完成时间
                            .setConnectTimeout(2000)
                            //请求超时,数据传输过程中数据包之间间隔的最大时间
                            .setSocketTimeout(3000)
                            //使用连接池来管理连接,从连接池获取连接的超时时间
                            .setConnectionRequestTimeout(20000)
                            .build();

                    //配置io线程
                    IOReactorConfig ioReactorConfig = IOReactorConfig.custom().
                            setIoThreadCount(Runtime.getRuntime().availableProcessors())
                            .setSoKeepAlive(true)
                            .build();
                    //设置连接池大小
                    ConnectingIOReactor ioReactor = null;
                    try {
                        ioReactor = new DefaultConnectingIOReactor(ioReactorConfig);
                    } catch (IOReactorException e) {
                        LOG.error("ioReactor连接池实例化失败", e);
                    }
                    PoolingNHttpClientConnectionManager connManager = new PoolingNHttpClientConnectionManager(ioReactor);
                    //最大连接数设置5
                    connManager.setMaxTotal(5);
                    //per route最大连接数设置
                    connManager.setDefaultMaxPerRoute(5);

                    client = HttpAsyncClients.custom()
                            .setConnectionManager(connManager)
                            .setDefaultRequestConfig(requestConfig)
                            .build();
                    client.start();

                }
            }
        }
        return client;
    }

    /**
     * 构建httpPost实体
     * @param url        请求地址
     * @param body       请求体body
     * @param contentType 编码方式
     * @param token       验证方式 token
     * @param sign        验证方式 sign
     * @return 构建好的httpPost实体
     */
    public static HttpPost getPostBody(String url, String body, String token, String sign, ContentType contentType) {
        HttpPost post;
        StringEntity entity;
        post = new HttpPost(url);
        if (!StringUtil.isEmpty(token)) {
            post.addHeader("token", token);
        }
        if (!StringUtil.isEmpty(sign)) {
            post.addHeader("sign", sign);
        }
        entity = new StringEntity(body, contentType);
        post.setEntity(entity);
        return post;
    }

    /**
     * 构建httpPost实体, contentType为 application/json utf-8
     * @param url 请求地址
     * @param body 请求体body
     * @param token       验证方式 token
     * @param sign        验证方式 sign
     * @return 构建好的httpPost实体
     */
    private static HttpPost getPostBody(String url, String body, String token, String sign) {
        return getPostBody(url, body, token, sign, ContentType.APPLICATION_JSON);
    }


    /**
     * 异步发送post请求
     * @param url       请求地址
     * @param body      请求体body
     * @param token     验证方式 token
     * @param sign      验证方式 sign
     */
    public static void executePost(String url, String body, String token, String sign) {
        CloseableHttpAsyncClient thisClient = getHttpClient();
        HttpPost post = getPostBody(url, body, token, sign);
        thisClient.execute(post, new FutureCallback<HttpResponse>() {
            @Override
            public void completed(HttpResponse httpResponse) {
                LOG.info("url:" + url + " body:" + body);
                try {
                    if (httpResponse.getStatusLine().getStatusCode() != 200) {
                        LOG.error("request url failed, http code=" + httpResponse.getStatusLine().getStatusCode() + ", url=" + url);
                    }
                    org.apache.http.HttpEntity entity = httpResponse.getEntity();
                    if (entity != null) {
                        String resultStr = EntityUtils.toString(entity, "utf-8");
                        LOG.info("url:" + url + " body:" + body + " resultStr = " + resultStr);
                    }
//                    LOG.info("{} -> {} completed", httpPost.getRequestLine(), httpResponse.getStatusLine());
                } catch (Exception e) {
                    LOG.error("executePost completed exception", e);
                }

            }

            @Override
            public void failed(Exception e) {
                LOG.error("executePost failed exception url:" + url + " body:" + body, e);
            }

            @Override
            public void cancelled() {
                LOG.error("executePost cancelled url:" + url + " body:" + body);
            }
        });
    }

    /**
     * 异步发送post请求(可重试)
     * 第一次失败后，最多重试两次
     * 第一次：0*1000
     * 第二次：1*1000
     * 第三次：2*1000
     * @param url 请求地址
     * @param body 请求体body
     * @param token 验证方式 token
     * @param sign 验证方式 sign
     */
    public static void executeRetryPost(String url, String body, String token, String sign) {
        executeRetryPost(url, body, token, sign, TRY_NUM);
    }

    /**
     * 异步发送post请求(可重试)
     * 第一次：0*1000
     * 第二次：1*1000
     * 第三次：2*1000
     * @param url       请求地址
     * @param body      请求体body
     * @param token     验证方式 token
     * @param sign      验证方式 sign
     * @param tryNum    尝试总次数
     */
    public static void executeRetryPost(String url, String body, String token, String sign, Integer tryNum) {
        if (Objects.isNull(tryNum)) {
            LOG.warn("tryNum is null！");
            return;
        }
        if (Objects.equals(tryNum, 0)) {
            LOG.warn("url：" + url + " 无尝试次数");
            return;
        }

        /**
         * 第一次：0*1000
         * 第二次：1*1000
         * 第三次：2*1000
         */
        trySleep(tryNum);

        if (Objects.nonNull(tryNum)) {
            String errorNumStr = staticPaasRedisHelper.getValue("errorNum", url);
            Integer errorNumInt = StringUtil.toInt(errorNumStr, 0);
            if (errorNumInt > 50) {
                LOG.warn("url:" + url +"在15分钟内连续失败50次，请在15分钟后再尝试！");
                return;
            }
        }
        CloseableHttpAsyncClient thisClient = getHttpClient();
        HttpPost post = getPostBody(url, body, token, sign);
        thisClient.execute(post, new FutureCallback<HttpResponse>() {
            @Override
            public void completed(HttpResponse httpResponse) {
                LOG.info("url:" + url + " body:" + body);
                try {
                    //请求成功时，清除错误次数计数缓存
                    staticPaasRedisHelper.removeValue("errorNum", url);
                    if (httpResponse.getStatusLine().getStatusCode() != 200) {
                        LOG.error("request url failed, http code=" + httpResponse.getStatusLine().getStatusCode() + ", url=" + url);
                    }
                    org.apache.http.HttpEntity entity = httpResponse.getEntity();
                    if (entity != null) {
                        String resultStr = EntityUtils.toString(entity, "utf-8");
                        LOG.info("resultStr = " + resultStr);
                    }
                } catch (Exception e) {
                    LOG.error("executePost completed exception", e);
                }
            }

            @Override
            public void failed(Exception e) {
                LOG.warn("failed retry num:" + (TRY_NUM - tryNum + 1) + " url:" + url + " body:" + body, e);
                //请求失败时，累加错误次数计数缓存
                staticPaasRedisHelper.getIncrNumByOne("errorNum", url, ERROR_NUM_SECONDS);
                if (Objects.nonNull(tryNum)) {
                    executeRetryPost(url, body, token, sign, tryNum-1);
                }
            }

            @Override
            public void cancelled() {
                LOG.info("failed retry num:" + (TRY_NUM - tryNum + 1) + " url:" + url + " body:" + body);
                //请求失败时，累加错误次数计数缓存
                staticPaasRedisHelper.getIncrNumByOne("errorNum", url, ERROR_NUM_SECONDS);
                if (Objects.isNull(tryNum)) {
                    executeRetryPost(url, body, token, sign, tryNum-1);
                }
            }
        });
    }

    /**
     *
     * @param tryNum
     */
    private static void trySleep(Integer tryNum) {
        long sleepMillis = (TRY_NUM-tryNum) * 1000;
        if (sleepMillis > 0) {
            try {
                Thread.sleep(sleepMillis);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}
